#include "utils/market_data.h"
#include <chrono>
#include <cstring> // 用于 strcpy
#include <fcntl.h>
#include <iostream>
#include <sys/mman.h>
#include <thread>
#include <unistd.h>

class DataCenter {
public:
  DataCenter() {
    // 创建共享内存（仅第一个生产者执行）
    bool is_first_producer = false;
    shm_fd = shm_open(SHM_NAME, O_CREAT | O_RDWR, 0666);
    if (shm_fd != -1) {
      is_first_producer = true;
      ftruncate(shm_fd, sizeof(ShmHeader) + BUFFER_CAPACITY * sizeof(Tick));
    }

    void *addr =
        mmap(nullptr, sizeof(ShmHeader) + BUFFER_CAPACITY * sizeof(Tick),
             PROT_READ | PROT_WRITE, MAP_SHARED, shm_fd, 0);
    header = reinterpret_cast<ShmHeader *>(addr);
    buffer = reinterpret_cast<Tick *>(header + 1);

    if (is_first_producer) {
      new (header) ShmHeader();
      header->capacity = BUFFER_CAPACITY;
      header->item_size = sizeof(Tick);
      sem_init(&header->data_ready, 1, 0);
      sem_init(&header->space_available, 1, BUFFER_CAPACITY);
    }
  }

  ~DataCenter() {
    // 仅最后一个生产者清理资源
    if (producer_count.fetch_sub(1) == 1) {
      sem_destroy(&header->data_ready);
      sem_destroy(&header->space_available);
      munmap(header, sizeof(ShmHeader) + BUFFER_CAPACITY * sizeof(Tick));
      close(shm_fd);
      shm_unlink(SHM_NAME);
    }
  }

  void StartProducer(const char *stock_symbol) {
    producer_count.fetch_add(1);
    std::thread(&DataCenter::ProduceTicks, this, stock_symbol).detach();
  }

private:
  void ProduceTicks(const char *stock_symbol) {
    uint64_t seq = 0;
    while (true) {
      Tick tick = GenerateTick(seq++, stock_symbol);

      // CAS预留写入位置
      size_t current_head = header->head.load(std::memory_order_relaxed);
      size_t new_head;
      do {
        new_head = current_head + 1;
        if (new_head - header->tail.load(std::memory_order_acquire) >
            BUFFER_CAPACITY) {
          std::this_thread::yield();
          continue;
        }
      } while (!header->head.compare_exchange_weak(current_head, new_head,
                                                   std::memory_order_acq_rel,
                                                   std::memory_order_relaxed));

      // 写入数据
      buffer[current_head % BUFFER_CAPACITY] = tick;

      // 顺序提交
      size_t expected_commit = current_head;
      while (!header->commit.compare_exchange_weak(expected_commit, new_head,
                                                   std::memory_order_release,
                                                   std::memory_order_relaxed)) {
        expected_commit = header->commit.load(std::memory_order_relaxed);
      }

      sem_post(&header->data_ready);
      std::this_thread::sleep_for(std::chrono::milliseconds(3000));
    }
  }

  Tick GenerateTick(uint64_t seq, const char *stock_symbol) {
    Tick tick;
    tick.timestamp = static_cast<uint64_t>(
        std::chrono::steady_clock::now().time_since_epoch().count());
    strncpy(tick.symbol, stock_symbol, sizeof(tick.symbol) - 1);
    tick.symbol[sizeof(tick.symbol) - 1] = '\0'; // 确保字符串以空字符结尾
    tick.price = 150.0 + (seq % 100) * 0.1;
    tick.volume = 100 * (seq % 10 + 1);
    return tick;
  }
  int shm_fd;
  ShmHeader *header;
  Tick *buffer;
  std::atomic<int> producer_count{0};
};

int main() {
  DataCenter dc1, dc2;
  dc1.StartProducer("MAOTAI");    // 启动第一个生产者
  dc2.StartProducer("WULIANGYE"); // 启动第二个生产者

  std::this_thread::sleep_for(std::chrono::seconds(300));
  return 0;
}